package com.nakanomay.zkcase2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * Created with IntelliJ IDEA.
 *
 * @author nakano_may丶
 * @date 2023/4/18
 * @Version 1.0
 * @description zk实现分布式锁🔐
 */
public class DistributedLock
{

    private String connectString = "172.16.102.100:2181,172.16.102.135:2181,172.16.102.136:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;
    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);
    private String waitPath;
    private String currentMode;

    public DistributedLock() throws IOException, InterruptedException, KeeperException
    {
        // 获取连接
        zk = new ZooKeeper(connectString, sessionTimeout, watchedEvent -> {
            // connectLatch 如果连接上zk 可以释放
            if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
                connectLatch.countDown();
            }

            // waitLatch 需要释放
            if (watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
                waitLatch.countDown();
            }
        });

        // 等待zk正常连接后，往下走程序
        connectLatch.await();

        // 判断根节点/locks是否存在？
        Stat stat = zk.exists("/locks", false);

        if (stat == null) {
            // 创建一下根节点
            zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    public void zkLock()
    {
        try {
            // 创建对应的临时带序号的节点
            currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // 判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听它序号前一个节点
            List<String> children = zk.getChildren("/locks", false);

            // 如果children 只有一个值，那么就直接获取锁，如果多有个节点，需要判断，谁最小
            if (children.size() == 1) {
                return;
            }
            else {
                Collections.sort(children);

                // 获取节点名称： seq-0000000
                String thisNode = currentMode.substring("/locks/".length());
                // 通过 seq-0000000 获取到节点在children中的位置
                int index = children.indexOf(thisNode);

                // 判断
                if (index == -1) {
                    System.out.println("数据异常");
                }
                else if (index == 0) {
                    // 就一个节点，获取到锁了
                    return;
                }
                else {
                    // 需要监听，前一个节点变化
                    waitPath = "/locks/" + children.get(index - 1);
                    zk.getData(waitPath, true, new Stat());

                    // 等待监听
                    waitLatch.await();
                    return;
                }
            }
        }
        catch (KeeperException e) {
            e.printStackTrace();
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 解锁方法
    public void unZkLock()
    {
        // 删除节点
        try {
            zk.delete(currentMode, -1);
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
        catch (KeeperException e) {
            e.printStackTrace();
        }
    }
}
